package beautiful.butterfly.drds.data_exchange.statistics.communication;

import beautiful.butterfly.drds.data_exchange.statistics.PerfTrace;
import beautiful.butterfly.drds.data_exchange.util.Strings;
import com.alibaba.fastjson.JSON;
import org.apache.commons.lang3.Validate;

import java.text.DecimalFormat;
import java.util.HashMap;
import java.util.Map;


/**
 * 这里主要是业务层面的处理
 */
public final class CommunicationTool
{
    public static final String stage = "stage";
    public static final String byte_speed = "byteSpeed";
    public static final String record_speed = "recordSpeed";
    public static final String percentage = "percentage";

    public static final String read_succeed_records = "readSucceedRecords";
    public static final String read_succeed_bytes = "readSucceedBytes";

    public static final String read_failed_records = "readFailedRecords";
    public static final String read_failed_bytes = "readFailedBytes";

    public static final String write_received_records = "writeReceivedRecords";
    public static final String write_received_bytes = "writeReceivedBytes";

    public static final String write_failed_records = "writeFailedRecords";
    public static final String write_failed_bytes = "writeFailedBytes";

    public static final String total_read_records = "totalReadRecords";
    public static final String wait_writer_time = "waitWriterTime";
    public static final String wait_reader_time = "waitReaderTime";
    public static final String transformer_used_time = "totalTransformerUsedTime";
    public static final String transformer_succeed_records = "totalTransformerSuccessRecords";
    public static final String transformer_failed_records = "totalTransformerFailedRecords";
    public static final String transformer_filter_records = "totalTransformerFilterRecords";
    public static final String transformer_name_prefix = "usedTimeByTransformer_";
    private static final String total_read_bytes = "totalReadBytes";
    private static final String total_error_records = "totalErrorRecords";
    private static final String total_error_bytes = "totalErrorBytes";
    private static final String write_succeed_records = "writeSucceedRecords";
    private static final String write_succeed_bytes = "writeSucceedBytes";

    public static Communication getReportCommunication(Communication now, Communication old, int totalStage)
    {
        Validate.isTrue(now != null && old != null,
                "为汇报准备的新旧metric不能为null");

        long totalReadRecords = getTotalReadRecords(now);
        long totalReadBytes = getTotalReadBytes(now);
        now.setLongCounter(total_read_records, totalReadRecords);
        now.setLongCounter(total_read_bytes, totalReadBytes);
        now.setLongCounter(total_error_records, getTotalErrorRecords(now));
        now.setLongCounter(total_error_bytes, getTotalErrorBytes(now));
        now.setLongCounter(write_succeed_records, getWriteSucceedRecords(now));
        now.setLongCounter(write_succeed_bytes, getWriteSucceedBytes(now));

        long timeInterval = now.getTimestamp() - old.getTimestamp();
        long sec = timeInterval <= 1000 ? 1 : timeInterval / 1000;
        long bytesSpeed = (totalReadBytes
                - getTotalReadBytes(old)) / sec;
        long recordsSpeed = (totalReadRecords
                - getTotalReadRecords(old)) / sec;

        now.setLongCounter(byte_speed, bytesSpeed < 0 ? 0 : bytesSpeed);
        now.setLongCounter(record_speed, recordsSpeed < 0 ? 0 : recordsSpeed);
        now.setDoubleCounter(percentage, now.getLongCounter(stage) / (double) totalStage);

        if (old.getThrowable() != null)
        {
            now.setThrowable(old.getThrowable());
        }

        return now;
    }

    public static long getTotalReadRecords(final Communication communication)
    {
        return communication.getLongCounter(read_succeed_records) +
                communication.getLongCounter(read_failed_records);
    }

    public static long getTotalReadBytes(final Communication communication)
    {
        return communication.getLongCounter(read_succeed_bytes) +
                communication.getLongCounter(read_failed_bytes);
    }

    public static long getTotalErrorRecords(final Communication communication)
    {
        return communication.getLongCounter(read_failed_records) +
                communication.getLongCounter(write_failed_records);
    }

    public static long getTotalErrorBytes(final Communication communication)
    {
        return communication.getLongCounter(read_failed_bytes) +
                communication.getLongCounter(write_failed_bytes);
    }

    public static long getWriteSucceedRecords(final Communication communication)
    {
        return communication.getLongCounter(write_received_records) -
                communication.getLongCounter(write_failed_records);
    }

    public static long getWriteSucceedBytes(final Communication communication)
    {
        return communication.getLongCounter(write_received_bytes) -
                communication.getLongCounter(write_failed_bytes);
    }

    public static class Stringify
    {
        private final static DecimalFormat df = new DecimalFormat("0.00");

        public static String getSnapshot(final Communication communication)
        {
            StringBuilder sb = new StringBuilder();
            sb.append("Total ");
            sb.append(getTotal(communication));
            sb.append(" | ");
            sb.append("Speed ");
            sb.append(getSpeed(communication));
            sb.append(" | ");
            sb.append("Error ");
            sb.append(getError(communication));
            sb.append(" | ");
            sb.append(" All task WaitWriterTime ");
            sb.append(PerfTrace.unitTime(communication.getLongCounter(wait_writer_time)));
            sb.append(" | ");
            sb.append(" All task WaitReaderTime ");
            sb.append(PerfTrace.unitTime(communication.getLongCounter(wait_reader_time)));
            sb.append(" | ");
            if (communication.getLongCounter(CommunicationTool.transformer_used_time) > 0
                    || communication.getLongCounter(CommunicationTool.transformer_succeed_records) > 0
                    || communication.getLongCounter(CommunicationTool.transformer_failed_records) > 0
                    || communication.getLongCounter(CommunicationTool.transformer_filter_records) > 0)
            {
                sb.append("Transfermor Success ");
                sb.append(String.format("%d records", communication.getLongCounter(CommunicationTool.transformer_succeed_records)));
                sb.append(" | ");
                sb.append("Transformer Error ");
                sb.append(String.format("%d records", communication.getLongCounter(CommunicationTool.transformer_failed_records)));
                sb.append(" | ");
                sb.append("Transformer Filter ");
                sb.append(String.format("%d records", communication.getLongCounter(CommunicationTool.transformer_filter_records)));
                sb.append(" | ");
                sb.append("Transformer usedTime ");
                sb.append(PerfTrace.unitTime(communication.getLongCounter(CommunicationTool.transformer_used_time)));
                sb.append(" | ");
            }
            sb.append("Percentage ");
            sb.append(getPercentage(communication));
            return sb.toString();
        }

        private static String getTotal(final Communication communication)
        {
            return String.format("%d records, %d bytes",
                    communication.getLongCounter(total_read_records),
                    communication.getLongCounter(total_read_bytes));
        }

        private static String getSpeed(final Communication communication)
        {
            return String.format("%s/s, %d records/s",
                    Strings.stringify(communication.getLongCounter(byte_speed)),
                    communication.getLongCounter(record_speed));
        }

        private static String getError(final Communication communication)
        {
            return String.format("%d records, %d bytes",
                    communication.getLongCounter(total_error_records),
                    communication.getLongCounter(total_error_bytes));
        }

        private static String getPercentage(final Communication communication)
        {
            return df.format(communication.getDoubleCounter(percentage) * 100) + "%";
        }
    }

    public static class Jsonify
    {
        @SuppressWarnings("rawtypes")
        public static String getSnapshot(Communication communication)
        {
            Validate.notNull(communication);

            Map<String, Object> state = new HashMap<String, Object>();

            Pair pair = getTotalBytes(communication);
            state.put((String) pair.getKey(), pair.getValue());

            pair = getTotalRecords(communication);
            state.put((String) pair.getKey(), pair.getValue());

            pair = getSpeedRecord(communication);
            state.put((String) pair.getKey(), pair.getValue());

            pair = getSpeedByte(communication);
            state.put((String) pair.getKey(), pair.getValue());

            pair = getStage(communication);
            state.put((String) pair.getKey(), pair.getValue());

            pair = getErrorRecords(communication);
            state.put((String) pair.getKey(), pair.getValue());

            pair = getErrorBytes(communication);
            state.put((String) pair.getKey(), pair.getValue());

            pair = getErrorMessage(communication);
            state.put((String) pair.getKey(), pair.getValue());

            pair = getPercentage(communication);
            state.put((String) pair.getKey(), pair.getValue());

            pair = getWaitReaderTime(communication);
            state.put((String) pair.getKey(), pair.getValue());

            pair = getWaitWriterTime(communication);
            state.put((String) pair.getKey(), pair.getValue());

            return JSON.toJSONString(state);
        }

        private static Pair<String, Long> getTotalBytes(final Communication communication)
        {
            return new Pair<String, Long>("totalBytes", communication.getLongCounter(total_read_bytes));
        }

        private static Pair<String, Long> getTotalRecords(final Communication communication)
        {
            return new Pair<String, Long>("totalRecords", communication.getLongCounter(total_read_records));
        }

        private static Pair<String, Long> getSpeedByte(final Communication communication)
        {
            return new Pair<String, Long>("speedBytes", communication.getLongCounter(byte_speed));
        }

        private static Pair<String, Long> getSpeedRecord(final Communication communication)
        {
            return new Pair<String, Long>("speedRecords", communication.getLongCounter(record_speed));
        }

        private static Pair<String, Long> getErrorRecords(final Communication communication)
        {
            return new Pair<String, Long>("errorRecords", communication.getLongCounter(total_error_records));
        }

        private static Pair<String, Long> getErrorBytes(final Communication communication)
        {
            return new Pair<String, Long>("errorBytes", communication.getLongCounter(total_error_bytes));
        }

        private static Pair<String, Long> getStage(final Communication communication)
        {
            return new Pair<String, Long>("stage", communication.getLongCounter(stage));
        }

        private static Pair<String, Double> getPercentage(final Communication communication)
        {
            return new Pair<String, Double>("percentage", communication.getDoubleCounter(percentage));
        }

        private static Pair<String, String> getErrorMessage(final Communication communication)
        {
            return new Pair<String, String>("errorMessage", communication.getThrowableMessage());
        }

        private static Pair<String, Long> getWaitReaderTime(final Communication communication)
        {
            return new Pair<String, Long>("waitReaderTime", communication.getLongCounter(CommunicationTool.wait_reader_time));
        }

        private static Pair<String, Long> getWaitWriterTime(final Communication communication)
        {
            return new Pair<String, Long>("waitWriterTime", communication.getLongCounter(CommunicationTool.wait_writer_time));
        }

        static class Pair<K, V>
        {
            private K key;
            private V value;

            public Pair(final K key, final V value)
            {
                this.key = key;
                this.value = value;
            }

            public K getKey()
            {
                return key;
            }

            public V getValue()
            {
                return value;
            }
        }
    }
}
